package com.desheng.bigdata.flink.stream.source

import java.io.IOException
import java.sql.{Connection, DriverManager, PreparedStatement}

import com.mysql.jdbc.Driver
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

/**
  * 自定义sourcefunction从mysql中读取数据
  * 非并行操作
  *     加载test库中的person表
  */
object _04UserDefineRichSourceFromMySQL {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val url = "jdbc:mysql://localhost:3306/test"
        val persons = env.addSource(new MySQLRichSourceFunction(url, user = "root", password = "sorry"))

        persons.print()

        env.execute(s"${_04UserDefineRichSourceFromMySQL.getClass.getSimpleName}")
    }
}

class MySQLRichSourceFunction(url: String, user: String, password: String) extends RichSourceFunction[Person] {

    classOf[Driver]
    var connection: Connection = null
    var ps: PreparedStatement = null
    override def open(parameters: Configuration): Unit = {
        connection = DriverManager.getConnection(url, user, password)
        val sql =
            """
              |select
              |  name,
              |  age,
              |  height,
              |  province
              |from person
              |where age > 13
            """.stripMargin
        ps = connection.prepareStatement(sql)
    }

    override def run(ctx: SourceFunction.SourceContext[Person]): Unit = {
        val rs = ps.executeQuery()
        while (rs.next()) {
            val name = rs.getString("name")
            val age = rs.getInt("age")
            val height = rs.getDouble("height")
            val province = rs.getString("province")
            val person = Person(name, age, height, province)
            ctx.collect(person)
            Thread.sleep(1000)
        }
    }


    override def close(): Unit = {
        try {
            if(ps != null) {
                ps.close()
            }
        } catch {
            case e: IOException => e.printStackTrace()
            case _ => {}
        } finally {
            try {
                if(connection != null) {
                    connection.close()
                }
            } catch {
                case e: IOException => e.printStackTrace()
                case _ => {}
            }
        }
    }
    override def cancel(): Unit = {

    }
}

case class Person(name: String, age: Int, height: Double, province: String)
